package com.wsjj.yjh;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class SQLDemo {


        public static void main(String[] args) {

            StreamExecutionEnvironment 	env 	= StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            StreamTableEnvironment 	tableEnv 	= StreamTableEnvironment.create(env);

            tableEnv.executeSql("CREATE TABLE flink_doris (\n" +
                 "    id int,\n" +
                            "    name varchar(50),\n" +
                            "    age int" +
                    "    ) \n" +
                    "    WITH (\n" +
                    "      'connector' = 'doris',\n" +
                    "      'fenodes' = '192.168.10.105:8030',\n" +
                    "      'table.identifier' = 'test_doris.student_kafka',\n" +
                    "      'username' = 'root',\n" +
                    "      'password' = '000000'\n" +
                    ")\n");


            // 读取数据
        tableEnv.executeSql("select * from flink_doris").print();

            // 写入数据
//            tableEnv.executeSql("insert 	into flink_doris(siteid,username,pv) values(22,'wuyanzu',3)");


        }

}
